package org.apache.spark.akka

import akka.actor.{Actor, ActorSelection, ActorSystem, Props}
import com.typesafe.config.ConfigFactory
import org.apache.spark.common.Constant
import org.apache.spark.common.Message.{Heartbeat, NotifyOffline, NotifyOnline, RegisterClient, RegisteredClient, SendMessage}

/**
 * @author: chenzhidiao
 * @date: 2021/1/4 10:36
 * @description:
 * @version: 1.0
 */
class MyClient(val clientHostname: String, clientid:String, val clientport:Int, val serverHostname: String, val serverPort: Int,ClientActorSystemName:String,ClinetActorName:String) extends Actor{
  var clientId: String = clientid
  var serverRef: ActorSelection = _

  override def preStart(): Unit = {

    //获取消息发送对象的一个ref实例
    // 远程path　akka.tcp://（ActorSystem的名称）@（远程地址的IP）:（远程地址的端口）/user/（Actor的名称）
    serverRef = context.actorSelection(s"akka.tcp://${
      Constant.SAS
    }@${serverHostname}:${serverPort}/user/${Constant.SA}")

    //发送消息
    println(clientHostname + " 正在注册")
    serverRef ! RegisterClient(clientid,clientHostname,clientport,ClientActorSystemName,ClinetActorName)
  }


  override def receive: Receive = {
    case RegisteredClient(serverHostname)=>{
      import scala.concurrent.duration._
      import context.dispatcher
      context.system.scheduler.schedule(0 millis, 4000 millis, self, SendMessage)
    }

    case SendMessage =>{
      serverRef ! Heartbeat(clientid)
      //println("线程id:"+Thread.currentThread().getId)
    }

    case NotifyOnline(clientid)=>{
      println(s"客户端${clientid}上线")
    }
    case NotifyOffline(clientid)=>{
      println(s"客户端${clientid}下线")
    }

  }
}

object  MyClient{
  //启动参数：1.客户端端口号；2.客户端id；3.客户端ActorSystem名称；4.客户端Actor名称
  //例如：7111 clinet1 MyClientActorSystem01 MyClientActor01
  def main(args: Array[String]): Unit = {
    var HostName = "localhost"

    var ServerHostName = "localhost"
    var ServerPort = 6789

    var ClientHostName = "localhost"
    var ClientPort = args(0).toInt
    var ClientId = args(1)
    println(ClientId)

    val str =
      s"""
         |akka.actor.provider = "akka.remote.RemoteActorRefProvider"
         |akka.remote.netty.tcp.hostname = ${ClientHostName}
         |akka.remote.netty.tcp.port = ${ClientPort}
      """.stripMargin
    val conf = ConfigFactory.parseString(str)

    //创建客户端的ActorSystem
    val actorSystem = ActorSystem(args(2), conf)

    //启动了一个actor ： MyClient
    actorSystem.actorOf(Props(new MyClient(ClientHostName, ClientId,ClientPort,ServerHostName,ServerPort,args(2),args(3))), args(3))
  }
}